package com.oops.core;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.oops.RabbitMQConstants;
import com.oops.model.bo.OrderBo;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Random;


@Component
@Slf4j
public class OrderConsumer {
    private Random random = new Random();

    @RabbitListener(queues = {RabbitMQConstants.ORDER_BIZ_QUEUE_NAME}, ackMode = "MANUAL", messageConverter = "jackson2JsonMessageConverter")
    public void consume(Channel channel, Message message, OrderBo orderBo,
                        @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        log.info("Thread.currentThread().getName()-> {}",Thread.currentThread().getName());
        log.info("message:{}", message);
        log.info("orderBo:{}", orderBo);
        try {
            if (random.nextInt(10) % 3 != 0) {
                // 手动拒绝消息
                log.info("basicReject deliveryTag:{}", deliveryTag);
                channel.basicReject(deliveryTag, false);
            } else {
                //第二个参数：是否批量处理
                final byte[] body = message.getBody();
                ObjectMapper objectMapper=new ObjectMapper();
                final OrderBo orderBo1 = objectMapper.readValue(new String(body, StandardCharsets.UTF_8), OrderBo.class);
                channel.basicAck(deliveryTag, false);
                log.info("basicAck deliveryTag:{} orderbo:{}", deliveryTag,orderBo1);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @RabbitListener(queues = {RabbitMQConstants.ORDER_DLX_QUEUE_NAME}, ackMode = "MANUAL", messageConverter = "jackson2JsonMessageConverter")
    public void dlxConsumer(Channel channel, Message message,
                            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        log.info("dlx message:{}", message);
        try {
            //第二个参数：是否批量处理
            log.info("dlx basicAck deliveryTag:{}", deliveryTag);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
